package org.xq.softcup.mq.remoting.invoker;

import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xq.softcup.mq.registry.ServiceRegistry;
import org.xq.softcup.mq.registry.impl.LocalServiceRegistry;
import org.xq.softcup.mq.registry.impl.XQRegistryServiceRegistry;
import org.xq.softcup.mq.remoting.net.params.BaseCallback;
import org.xq.softcup.mq.remoting.net.params.XQRpcFutureResponse;
import org.xq.softcup.mq.remoting.net.params.XQRpcResponse;
import org.xq.softcup.mq.util.ThreadPoolUtil;
import org.xq.softcup.mq.util.XQRpcException;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
public class XQRpcInvokerFactory {
    //LocalServiceRegistry.class
    private static volatile XQRpcInvokerFactory instance = new XQRpcInvokerFactory(XQRegistryServiceRegistry.class, null);
    private Class<? extends ServiceRegistry> serviceRegistryClass;
    private Map<String, String> serviceRegistryParam;
    private ServiceRegistry serviceRegistry;

    public static XQRpcInvokerFactory getInstance() {
        return instance;
    }

    public XQRpcInvokerFactory() {
    }

    public XQRpcInvokerFactory(Class<? extends ServiceRegistry> serviceRegistryClass, Map<String, String> serviceRegistryParam) {
        this.serviceRegistryClass = serviceRegistryClass;
        this.serviceRegistryParam = serviceRegistryParam;
    }

    public void start() throws Exception {
        if (this.serviceRegistryClass != null) {
            this.serviceRegistry = ((ServiceRegistry) this.serviceRegistryClass.newInstance());
            this.serviceRegistry.start(this.serviceRegistryParam);
        }
    }

    public void stop() throws Exception {
        if (this.serviceRegistry != null) {
            this.serviceRegistry.stop();
        }
        if (this.stopCallbackList.size() > 0) {
            for (BaseCallback callback : this.stopCallbackList) {
                try {
                    callback.run();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }
        stopCallbackThreadPool();
    }

    public ServiceRegistry getServiceRegistry() {
        return this.serviceRegistry;
    }

    private List<BaseCallback> stopCallbackList = new ArrayList();

    public void addStopCallBack(BaseCallback callback) {
        this.stopCallbackList.add(callback);
    }

    private ConcurrentMap<String, XQRpcFutureResponse> futureResponsePool = new ConcurrentHashMap();

    public void setInvokerFuture(String requestId, XQRpcFutureResponse futureResponse) {
        this.futureResponsePool.put(requestId, futureResponse);
    }

    public void removeInvokerFuture(String requestId) {
        this.futureResponsePool.remove(requestId);
    }

    public void notifyInvokerFuture(String requestId, final XQRpcResponse xqRpcResponse) {
        final XQRpcFutureResponse futureResponse = (XQRpcFutureResponse) this.futureResponsePool.get(requestId);
        if (futureResponse == null) {
            return;
        }
        if (futureResponse.getInvokeCallback() != null) {
            try {
                executeResponseCallback(new Runnable() {
                    public void run() {
                        if (xqRpcResponse.getErrorMsg() != null) {
                            futureResponse.getInvokeCallback().onFailure(new XQRpcException(xqRpcResponse.getErrorMsg()));
                        } else {
                            futureResponse.getInvokeCallback().onSuccess(xqRpcResponse.getResult());
                        }
                    }
                });
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        } else {
            futureResponse.setResponse(xqRpcResponse);
        }
        this.futureResponsePool.remove(requestId);
    }

    private ThreadPoolExecutor responseCallbackThreadPool = null;

    public void executeResponseCallback(Runnable runnable) {
        if (this.responseCallbackThreadPool == null) {
            synchronized (this) {
                if (this.responseCallbackThreadPool == null) {
                    this.responseCallbackThreadPool = ThreadPoolUtil.makeServerThreadPool(XQRpcInvokerFactory.class.getSimpleName());
                }
            }
        }
        this.responseCallbackThreadPool.execute(runnable);
    }

    public void stopCallbackThreadPool() {
        if (this.responseCallbackThreadPool != null) {
            this.responseCallbackThreadPool.shutdown();
        }
    }
}
